package com.zipking.cloud.springbootmybatis.util;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.zipking.cloud.springbootmybatis.persistance.entity.MsgLog;
import lombok.extern.slf4j.Slf4j;

import javax.naming.ServiceUnavailableException;
import java.util.*;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * @author wangguangcai
 * @version 1.0
 * @ClassName Batch
 * Description: 批处理相关操作
 * @date 2024/4/1 9:04
 */
@Slf4j
public class BatchUtils {
    /**
     * 批次数量
     */
    public static final Integer NUMBER_BACH_PROTECT = 100;


    /**
     * 批量中包含有关联查询，可能因为数据过长出现数据查询的的问题，
     * 这里进行分批处理
     *
     * @param list 需要处理的list
     * @param bach 批量处理的函数
     * @param <T>  元素类型
     */
    public static <T> void protectBach(List<T> list, Consumer<List<T>> bach) {
        if (isEmpty(list)) {
            return;
        }
        if (list.size() > NUMBER_BACH_PROTECT) {
            for (int i = 0; i < list.size(); i += NUMBER_BACH_PROTECT) {
                int lastIndex = Math.min(i + NUMBER_BACH_PROTECT, list.size());
                bach.accept(list.subList(i, lastIndex));
            }
        } else {
            bach.accept(list);
        }
    }

    /**
     * 批量中包含有关联查询，可能因为数据过长出现数据查询的的问题
     * 这里进行分批，合并数据处理。
     *
     * @param list 原数据
     * @param bach 处理方法
     * @param <T>  原数据类型
     * @param <R>  处理结果类型
     * @return 处理结果汇总
     */
    public static <T, R> List<R> protectBach(List<T> list, Function<List<T>, List<R>> bach) {
        if (isEmpty(list)) {
            return Collections.emptyList();
        }
        if (list.size() > NUMBER_BACH_PROTECT) {
            List<R> end = new LinkedList<>();
            for (int i = 0; i < list.size(); i += NUMBER_BACH_PROTECT) {
                int lastIndex = Math.min(i + NUMBER_BACH_PROTECT, list.size());
                Optional.ofNullable(bach.apply(list.subList(i, lastIndex)))
                        .ifPresent(end::addAll);
            }
            return end;
        }
        return bach.apply(list);
    }

    private static <T> boolean isEmpty(Collection<T> collection) {
        return collection == null || collection.isEmpty();
    }

    /**
     * 批量操作
     *
     * @param pageSize
     * @param list
     * @param callback 要执行的方法
     * @param <T> 返回的数据
     * @return
     * @throws ServiceException
     */
    public static <T> int operate(int pageSize, List<T> list, Callback callback) throws ServiceUnavailableException {

        int total = list.size();
        int pages = total <= pageSize ? 1 : total % pageSize == 0 ? total / pageSize : total / pageSize + 1;
        try {
            for (int i = 0; i < pages; i++) {
                if ((i + 1) * pageSize <= total) {
                    callback.execute(list.subList(i * pageSize, (i + 1) * pageSize));
                } else {
                    callback.execute(list.subList(i * pageSize, total));
                }
            }
            return 1;
        } catch (Exception e) {
            log.error("处理失败:{}", e.getMessage());
        }
        return 0;
    }


    public static interface Callback<T> {
        void execute(List<T> list);
    }
    /**
     * list 有返回值
     * @param list     数据源
     * @param function
     * @param <T>      参数类型
     * @param <V>      返回类型
     * @return
     */
    /**
     * list 分成多个子list
     * @param list
     * @param pageCount 总页数 ，如果list条数不够，根据list条数来
     * @param <T>
     * @return
     */
    public static <T> List<List<T>> listToPage(List<T> list, int pageCount) {

        int currentCount = list.size();
        //总页数 真实线程数
        int page = pageCount >= currentCount ? currentCount : pageCount;
        //每页条数
        int pageSize = currentCount / page;

        List<List<T>> pageList = new ArrayList<>();
        for (int i = 0; i < page; i++) {

            //采用流方式分页
            List<T> lastPage = list.stream().
                    skip(i * pageSize).
                    limit(pageSize).
                    collect(Collectors.toList());

            if (page * pageSize + i < currentCount) {
                lastPage.add(
                        list.stream().
                                skip(page * pageSize + i).
                                limit(1).
                                findFirst().
                                get()
                );
            }
            pageList.add(lastPage);
        }
        return pageList;
    }


    /**
     * 分批执行方法 采用：Executors+分页方法
     * @param list           数据源
     * @param function       处理方法 lamda表达式
     * @param maxThreadCount 处理线程数量
     * @param <T>            参数类型
     * @param <V>            返回类型
     * @return 处理完成结果集
     */
    public static <T, V> List<V> listToExecutorPageSubmits(List<T> list, Function<T, V> function, int maxThreadCount) {
        List<V> results = new ArrayList<>();

        if (list == null || list.isEmpty()) {
            return results;
        }

        maxThreadCount = maxThreadCount <= 0 ? 1 : maxThreadCount;
        maxThreadCount = maxThreadCount >= 10 ? 10 : maxThreadCount;

        //分批数据源
        List<List<T>> pageList = listToPage(list, maxThreadCount);

        //分批并行执行
        List<Future<List<V>>> futures = new ArrayList<>();
        ExecutorService executorService = Executors.newFixedThreadPool(pageList.size());
        pageList.forEach(items -> {
            futures.add(
                    executorService.submit(() -> {
                        List<V> childResults = new ArrayList<>();
                        items.forEach(item -> childResults.add(function.apply(item)));
                        return childResults;
                    })
            );
        });

        //合并结果
        futures.forEach(b -> {
            try {
                results.addAll(b.get());
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (ExecutionException e) {
                e.printStackTrace();
            }
        });

        return results;
    }


    public static void main(String[] args) {
/*  用法1
        LambdaQueryWrapper<FeatureClazzEntity> wrapper=new LambdaQueryWrapper();
        wrapper.select(FeatureClazzEntity::getId).in(FeatureClazzEntity::getFeatureSetId,setIds);
        List<Long> ids=list(wrapper).stream().map(FeatureClazzEntity::getId).collect(Collectors.toList());
        BatchUtil.protectBach(ids, this::delete);*/

            //用法2
      /*   LambdaQueryWrapper<MsgLog> wrapper=new LambdaQueryWrapper();

        List<Long> ids=this.(wrapper).stream().map(FeatureClazzEntity::getId).collect(Collectors.toList());
        List<AttributeSetEntity> result= BatchUtils.protectBach(ids, idList->{
            return attributeSetService.lambdaQuery().in(AttributeSetEntity::getClazzId,idList).list();
        });*/


    }
}
